package bigdata.backend.monitor.binlog.controller;

import bigdata.backend.monitor.binlog.service.KafkaMonitorService;
import bigdata.backend.monitor.utils.KafkaProgressUtils;
import bigdata.backend.utils.Result;
import bigdata.backend.utils.ResultCodeEnum;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * Kafka 消费进度监控 Controller
 * 提供查询不同 Topic 消费进度的 API 接口
 */
@RestController
@RequestMapping("/api/monitor/kafka")
public class KafkaMonitorController {

    @Autowired
    private KafkaMonitorService kafkaMonitorService;

    /**
     * 查询 binlog 监控 Topic 的消费进度
     * 该 Topic 由 binlog-monitor-group 消费
     *
     * @return Result 包含 Topic 进度信息
     */
    @GetMapping("/binlog-progress")
    public Result<KafkaProgressUtils.TopicProgressInfo> getBinlogProgress() {
        try {
            KafkaProgressUtils.TopicProgressInfo progress = kafkaMonitorService.getBinlogTopicProgress();
            if (progress.getError() != null) {
                // 如果查询过程中发生错误，返回错误信息
                return Result.build(progress, ResultCodeEnum.INTERNAL_ERROR).message("查询 binlog 进度失败: " + progress.getError());
            }
            return Result.ok(progress);
        } catch (Exception e) {
            System.err.println("Error querying binlog progress: " + e.getMessage());
            e.printStackTrace();
            return Result.<KafkaProgressUtils.TopicProgressInfo>build(null, ResultCodeEnum.INTERNAL_ERROR).message("查询 binlog 进度时发生未知错误: " + e.getMessage());
        }
    }

    /**
     * 查询违规事件 Topic 的消费进度
     * 该 Topic 由 monitor-violation-events-group 消费
     *
     * @return Result 包含 Topic 进度信息
     */
    @GetMapping("/violation-progress")
    public Result<KafkaProgressUtils.TopicProgressInfo> getViolationProgress() {
        try {
            KafkaProgressUtils.TopicProgressInfo progress = kafkaMonitorService.getViolationTopicProgress();
            if (progress.getError() != null) {
                // 如果查询过程中发生错误，返回错误信息
                return Result.build(progress, ResultCodeEnum.INTERNAL_ERROR).message("查询违规事件进度失败: " + progress.getError());
            }
            return Result.ok(progress);
        } catch (Exception e) {
            System.err.println("Error querying violation progress: " + e.getMessage());
            e.printStackTrace();
            return Result.<KafkaProgressUtils.TopicProgressInfo>build(null, ResultCodeEnum.INTERNAL_ERROR).message("查询违规事件进度时发生未知错误: " + e.getMessage());
        }
    }

    // 如果需要，可以添加一个通用的查询接口，允许前端指定 Topic 和 GroupId
     @GetMapping("/progress")
     public Result<KafkaProgressUtils.TopicProgressInfo> getProgress(@RequestParam String topicName, @RequestParam String consumerGroupId) {
         try {
             KafkaProgressUtils.TopicProgressInfo progress = kafkaMonitorService.getTopicProgress(topicName, consumerGroupId);
             if (progress.getError() != null) {
                 return Result.build(progress, ResultCodeEnum.INTERNAL_ERROR).message("查询进度失败: " + progress.getError());
             }
             return Result.ok(progress);
         } catch (Exception e) {
             System.err.println("Error querying progress for topic " + topicName + " and group " + consumerGroupId + ": " + e.getMessage());
             e.printStackTrace();
             return Result.<KafkaProgressUtils.TopicProgressInfo>build(null, ResultCodeEnum.INTERNAL_ERROR).message("查询进度时发生未知错误: " + e.getMessage());
         }
     }
}
